package com.csdj.point.mq;

import com.csdj.beans.entity.Order;
import com.csdj.beans.interfaces.goods.IGoodsService;
import com.csdj.beans.interfaces.order.IOrderService;
import com.csdj.common.constant.CommonConstant;
import com.csdj.common.constant.OrderConstant;
import com.csdj.common.constant.PaymentConstant;
import com.csdj.point.service.PointService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;

/**
 * MessageListenerCurrently并行消费消息处理器 会用多个线程去消费同一个消息队列 ， 所以不能保证消息的消费顺序
 * 如果要保证消费消息顺序 请使用MessageListenerOrderly
 */
@Service
public class RocketMsgHandler implements MessageListenerOrderly {

    private Logger log= LoggerFactory.getLogger("point msg handler");
    @Resource
    private PointService pointService;

    private ObjectMapper om = new ObjectMapper();


    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        if(msgs != null){
            try{
                log.info("msg size:{}",msgs.size());
                msgs.stream().forEach(msg ->{
                    log.info("msg topic:{}  tags:{}",msg.getTopic(),msg.getTags());
                    if(PaymentConstant.PAYMENT_TAG.equals(msg.getTags())){
                        try {
                            Order order = om.readValue(msg.getBody(),Order.class);
                            pointService.updateByOrder(order,(byte)2);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
